package State

import Source.SensorReading
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object StateTest02 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    val alertStream = dataStream
      .keyBy(_.id)
      .flatMapWithState[(String, Double, Double), Double]({
        //(输入的数据类型，当前的状态类型) =>(,状态值的更改)=>
        case (data: SensorReading, None) => (List.empty, Some(data.temperature))
        case (data: SensorReading, lastTemp: Some[Double]) => {
          if ((data.temperature - lastTemp.get).abs >= 10)
            (List(("温差超过十度", lastTemp.get, data.temperature)), Some(data.temperature))
          else
            (List(("温度正常", lastTemp.get, data.temperature)), Some(data.temperature))
        }
      })

    alertStream.print()

    env.execute("TempChange")
  }
}
